package tbdp.tool.writer;

import org.apache.commons.io.IOUtils;

import java.io.FileInputStream;
import java.io.InputStream;
import java.io.OutputStream;

import alluxio.AlluxioURI;
import alluxio.client.ReadType;
import alluxio.client.WriteType;
import alluxio.client.file.FileSystem;
import alluxio.client.file.options.CreateFileOptions;
import alluxio.client.file.options.OpenFileOptions;

/**
 * alluxio writer
 *
 * @author xbkaishui
 * @version $Id: AlluxioWriter.java, v 0.1 2017-06-18 下午12:31 xbkaishui Exp $$
 */
public class AlluxioWriter extends HdfsWriter {

    private static OpenFileOptions mReadOptions;
    private static CreateFileOptions mWriteOptions;

    public AlluxioWriter() {
        mReadOptions = OpenFileOptions.defaults().setReadType(ReadType.CACHE_PROMOTE);
        mWriteOptions = CreateFileOptions.defaults().setWriteType(WriteType.CACHE_THROUGH);
    }

    /**
     * 写入到 alluxio 目录
     */
    @Override
    protected void saveFileToHdfs(String tmpFilePath, String path) {
        logger.info("save local file {}   to alluxio {} ", tmpFilePath, path);
        FileSystem fileSystem = FileSystem.Factory.get();
//        AlluxioURI mFilePath = new AlluxioURI("hdfs:/" + path);
        AlluxioURI mFilePath = new AlluxioURI(path);
        OutputStream oos = null;
        InputStream ins = null;
        try {
            oos = fileSystem.createFile(mFilePath, mWriteOptions);
            ins = new FileInputStream(tmpFilePath);
            IOUtils.copy(ins, oos);
        } catch (Throwable e) {
            throw new RuntimeException(e);
        } finally {
            IOUtils.closeQuietly(oos);
            IOUtils.closeQuietly(ins);
        }
    }

    @Override
    public String getType() {
        return "alluxio";
    }
}
